/**
 * 
 */
package com.ailk.bigdata.etl.realstream.server.service.impl;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.io.IOUtil;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * 
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2016-2-25]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class RollingFileWriter
{
	private static final Logger logger = LoggerFactory.getLogger(RollingFileWriter.class);
	private static final String inUseSuffix = ".tmp";
	private static final int fileFlagNum = 99;
	private static final int wirterSize = 4;

	private final int rollInterval;

	private volatile boolean shouldRotate = false;

	private List<BufferedWritter> fileWriters;
	private ScheduledExecutorService rollService;

	private AtomicInteger fileExtensionCounter;
	private AtomicInteger writeCount;

	private volatile String filePath;
	private volatile String fileName;

	protected RollingFileWriter(int rollInterval, String filePath, String fileName)
	{
		this.rollInterval = rollInterval;
		this.filePath = filePath;
		this.fileName = fileName;

		fileExtensionCounter = new AtomicInteger(0);
		writeCount = new AtomicInteger(0);

	}

	public void initWriterStreams() throws Exception
	{
		fileWriters = new ArrayList<BufferedWritter>();
		for (int i = 0; i < wirterSize; i++)
		{
			String fileName = getCurrentFileName();
			BufferedWritter bufferedWritter = new BufferedWritter(new OutputStreamWriter(new FileOutputStream(fileName), "UTF-8"), 2 * 1024 * 1024);
			bufferedWritter.setFileName(fileName);
			fileWriters.add(bufferedWritter);
		}

	}

	/**
	 * @description 往落地处理线程放入消息
	 * @param realData
	 * @throws IOException
	 */
	public void writeStream(String realData) throws Exception
	{
		if (shouldRotate)
		{
			synchronized (RollingFileWriter.class)
			{
				if (shouldRotate)
				{
					for (BufferedWritter fileWriter : fileWriters)
					{
						String fileName = fileWriter.getFileName();
						IOUtil.closeQuietly(fileWriter);
						rename(fileName);
					}

					initWriterStreams();

					writeCount.set(0);

					shouldRotate = false;
				}
			}
		}

		fileWriters.get(writeCount.incrementAndGet() % wirterSize).write(realData);

	}

	public void initRollIntervalTh()
	{
		// 如果按照时间间隔分批文件
		if (rollInterval > 0)
		{
			logger.info("处理文件定时分批线程启动");
			rollService = Executors.newScheduledThreadPool(1,
					new ThreadFactoryBuilder().setNameFormat("rollingFileSink-roller-" + Thread.currentThread().getId() + "-%d").build());

			rollService.scheduleAtFixedRate(new Runnable()
			{

				@Override
				public void run()
				{
					shouldRotate = true;
				}

			}, rollInterval, rollInterval, TimeUnit.MINUTES);
		}
	}

	/**
	 * @description 获取当前文件名称
	 */
	private String getCurrentFileName()
	{
		String fullFilePath = FilePathFormatUtil.escapeString(filePath, true, Calendar.HOUR_OF_DAY, 1);
		String fullFileName = FilePathFormatUtil.escapeString(FilenameUtils.getBaseName(fileName), true, Calendar.MINUTE, rollInterval)
				+ FilenameUtils.EXTENSION_SEPARATOR_STR + FilenameUtils.getExtension(fileName);

		int counter = fileExtensionCounter.incrementAndGet() % fileFlagNum;

		StringBuilder bucketPath = new StringBuilder(fullFilePath);
		bucketPath.append(File.separator).append(String.format(fullFileName, counter)).append(inUseSuffix);
		logger.info("创建接口落地文件路径[" + bucketPath + "]......");

		fullFilePath = null;
		fullFileName = null;

		return bucketPath.toString();
	}

	/**
	 * @description 重命名文件，去除临时文件标示
	 * @param bucketPath
	 */
	private void rename(String bucketPath)
	{
		File writedFile = new File(bucketPath);
		writedFile.renameTo(new File(StringUtils.removeEnd(bucketPath, inUseSuffix)));
	}
}

class BufferedWritter extends BufferedWriter
{
	private static String lineSeparator = System.getProperty("line.separator");
	private String fileName;

	@Override
	public void write(String str) throws IOException
	{
		super.write(str + lineSeparator);
	}

	/**
	 * @param out
	 * @param sz
	 */
	public BufferedWritter(Writer out, int sz)
	{
		super(out, sz);
	}

	/**
	 * @return the fileName
	 */
	public String getFileName()
	{
		return fileName;
	}

	/**
	 * @param fileName
	 *            the fileName to set
	 */
	public void setFileName(String fileName)
	{
		this.fileName = fileName;
	}

}
